package org.xserver.component.handler;

import java.lang.reflect.Method;

import javax.annotation.Resource;

import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.xserver.common.util.MessageUtils;
import org.xserver.common.util.StringUtil;
import org.xserver.component.config.XServerHttpConfig;
import org.xserver.component.core.dispatch.RequestDispatchFilter;
import org.xserver.component.core.dispatch.RequestDispatchManager;
import org.xserver.component.core.websocket.KeepAliveWebsocketResolver;
import org.xserver.component.core.websocket.WSMessage;
import org.xserver.component.core.websocket.WebSocketCacheManager;
import org.xserver.component.core.websocket.WebSocketClient;
import org.xserver.component.core.websocket.WebSocketResolverManager;
import org.xserver.component.exception.TokenFilterError;
import org.xserver.component.handler.ext.WriteHandlerFactory;
import org.xserver.component.json.JsonManager;
import org.xserver.component.redis.RedisTemplate;
import org.xserver.component.redis.util.RedisConstant;
import org.xserver.component.spring.SpringUtil;
import org.xserver.disruptor.core.DisruptorEventFactory;
import org.xserver.disruptor.producer.WebsocketEvent;


/**
 * Handle webSocket
 * 
 * @author idoldefine
 * @since 2014/04/22
 */
@Component
public class WebSocketHandler extends SimpleChannelUpstreamHandler {
	public static final Logger logger = LoggerFactory
			.getLogger(WebSocketHandler.class);
	
	@Resource
	private RedisTemplate redisTemplate;
	@Resource
	private RequestDispatchFilter requestDispatchFilter;

	@Resource
	private RequestDispatchManager requestDispatchManager;
	@Resource
	private WriteHandlerFactory writeHandlerFactory;
	@Resource
	private XServerHttpConfig httpConfig;
	@Resource
	private KeepAliveWebsocketResolver keepAliveWebsocketResolver;
	
	@Resource
	private DisruptorEventFactory disruptorEventFactory;
	
	private static final Logger log = LoggerFactory
			.getLogger(WebSocketHandler.class);

	public static final String WEBSOCKET = "websocket";

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		Object message = e.getMessage();
		if (message instanceof WebSocketFrame) {
			handleWebSocketFrame(ctx, e);
		}
		ctx.sendUpstream(e);
	}

	/**
	 * There will deal three {@link WebSocketFrame},
	 * <ul>
	 * <li>{@link CloseWebSocketFrame} that we just close the channel;</li>
	 * <li>{@link PingWebSocketFrame} that we return binary data;</li>
	 * <li>{@link TextWebSocketFrame} that we dispatch to business executor;</li>
	 * </ul>
	 * 
	 * @param ctx
	 * @param e
	 * @throws Exception 
	 */
	private void handleWebSocketFrame(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
		Object frame = e.getMessage();
		WebSocketCacheManager cacheManager = (WebSocketCacheManager) SpringUtil.getBean(WebSocketCacheManager.class);
        String id = ctx.getChannel().getId().toString();
        WebSocketClient client = cacheManager.getWebSocketClient(id);
		keepAliveWebsocketResolver.doHandleRequest(ctx, "", client);
		if (frame instanceof CloseWebSocketFrame) {
			ChannelFuture f = ctx.getChannel().write(frame);
			f.addListener(ChannelFutureListener.CLOSE);
			log.info("client {} close connection", e.getChannel()
					.getRemoteAddress());
			return;
		}

		if (frame instanceof PingWebSocketFrame) {
			ctx.getChannel().write(
					new PongWebSocketFrame(((PingWebSocketFrame) frame)
							.getBinaryData()));
			return;
		}
		if (frame instanceof PongWebSocketFrame) {
			
			return;
		}

		if (!(frame instanceof TextWebSocketFrame)) {
			log.error("{} frame types not supported", frame.getClass()
					.getName());
			throw new UnsupportedOperationException(String.format(
					"%s frame types not supported", frame.getClass().getName()));
		}
		TextWebSocketFrame webSocketFrame = (TextWebSocketFrame) frame;
		String text =  webSocketFrame.getText();
		if (text != null && text.length() > 0) {
			WSMessage message = MessageUtils.parseText(text);
			String token = message.getToken();

			if (message.getTopic() != null) {
				if (!message.getContentType().equals("login")
						&& StringUtil.isEmpty(token)) {
					MessageUtils.sendTokenErrorMessage(client);
					return;
				}
				if (!message.getContentType().equals("login")) {
					Integer ret = (Integer) redisTemplate.doValidate(token);
					if (!message.getContentType().equals("login")
							&& ret.intValue() != RedisConstant.TOKEN_SUCCESS
									.intValue()) {
						MessageUtils.sendTokenErrorMessage(client);
						return;
					}
				}

				Class<?> clazz = ((WebSocketResolverManager) SpringUtil
						.getBean(WebSocketResolverManager.class))
						.getInstance(message.getTopic());
				Method method = ((WebSocketResolverManager) SpringUtil
						.getBean(WebSocketResolverManager.class))
						.getMethod(message.getContentType());
				if (clazz == null || method == null) {
					return;
				}

				disruptorEventFactory.post(WebsocketEvent.getInstance(method,
						clazz, message, ctx.getChannel().getId().toString()));
			} else {
				if ("ack".equals(message.getContentType())) {
					// ack 计算客户端延时
					Long messageId = message.getId();
					WebSocketClient webSocketClient = cacheManager
							.getWebSocketClient(id);
					if (webSocketClient != null) {
						webSocketClient.setMessageId(messageId);
						webSocketClient.newLastAckTime();
					}

				}
			}

		}
	}

}
